package net.jueb.frameworkDemo.mq.rabbitMq;
import java.io.IOException;

import com.rabbitmq.client.BlockedListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;  
  
public class Recv  
{  
    //队列名称  
    private final static String QUEUE_NAME = "hello";  
  
    
    public void start() throws Exception
    {
    	  
        //打开连接和创建频道，与发送端一样  
        ConnectionFactory factory = new ConnectionFactory();  
        String uri="amqp://admin:admin@192.168.0.205:5672";//uri配置http://www.rabbitmq.com/uri-spec.html
//        factory.setHost("192.168.0.205");  
//        factory.setPort(5672);
//        factory.setUsername("admin");
//        factory.setPassword("admin");
        factory.setUri(uri);
        factory.setConnectionTimeout(1000);//配置链接超时
        factory.setAutomaticRecoveryEnabled(true);//开启重连
        factory.setNetworkRecoveryInterval(5000);//重连间隔
        Connection connection = factory.newConnection(); 
        connection.addBlockedListener(new BlockedListener() {
			
			@Override
			public void handleUnblocked() throws IOException {
				System.err.println("handleUnblocked");
			}
			
			@Override
			public void handleBlocked(String reason) throws IOException {
				System.err.println("handleBlocked:"+reason);
			}
		});
        connection.addShutdownListener(new ShutdownListener() {
			
			@Override
			public void shutdownCompleted(ShutdownSignalException cause) {
				System.err.println("connection shutdownCompleted:"+cause.getMessage());
			}
		});
        Channel channel = connection.createChannel();  
        channel.addShutdownListener(new ShutdownListener() {
			
			@Override
			public void shutdownCompleted(ShutdownSignalException cause) {
				System.err.println("channel shutdownCompleted:"+cause.getMessage());
			}
		});
        /**
         * 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。  
         * 第一个参数：队列名字，
         * 第二个参数：队列是否可持久化即重启后该队列是否依然存在，
         * 第三个参数：该队列是否时独占的即连接上来时它占用整个网络连接，
         * 第四个参数：是否自动销毁即当这个队列不再被使用的时候即没有消费者对接上来时自动删除，
         * 第五个参数：其他参数如TTL（队列存活时间）等。
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
        autoAck(channel);
    }
    
    enum AckResult {
    	/**
    	 * 处理
    	 */
    	ACCEPT,
    	/**
    	 * 重试
    	 */
    	RETRY,
    	/**
    	 * 丢弃
    	 */
    	REJECT,
    	}
    
    
    /**
     * 自动确认删除
     * @param channel
     * @throws IOException
     */
    public void autoAck(Channel channel) throws IOException
    {
       //创建队列消费者  
       QueueingConsumer consumer = new QueueingConsumer(channel);  
       //指定消费队列 (自动ack) 
       channel.basicConsume(QUEUE_NAME, true, consumer);  
       while(true)
       {
    	   try {
       		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");  
          	 //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）  
              QueueingConsumer.Delivery delivery = consumer.nextDelivery(1000);
              if(delivery!=null)
              {
              		String message = new String(delivery.getBody());  
              		System.out.println(" [x] Received '" + message + "'");  
              		Thread.sleep(1000);
              }
			} catch (Exception e) {
				System.out.println(e.getMessage());
			}
       }
    }
    
    /**
     * 手动确认删除
     * BasicReject方法第一个参数是消息的DeliveryTag，对于每个Channel来说，每个消息都会有一个DeliveryTag，一般用接收消息的顺序来表示：1,2,3,4 等等。第二个参数是是否放回queue中，requeue。
	 * BasicReject一次只能拒绝接收一个消息，而BasicNack方法可以支持一次0个或多个消息的拒收，并且也可以设置是否requeue。
     * @param channel
     * @throws IOException
     */
    public void manualAck(Channel channel) throws IOException
    {
    	 //创建队列消费者  
        QueueingConsumer consumer = new QueueingConsumer(channel);  
        //指定消费队列  第一个参数：队列名字，第二个参数：是否自动应答，如果为真，消息一旦被消费者收到，服务端就知道该消息已经投递，从而从队列中将消息剔除，否则，需要在消费者端手工调用
        channel.basicConsume(QUEUE_NAME, true, consumer);  
        while(true)
        {
    	   try {
       		  System.out.println(" [*] Waiting for messages. To exit press CTRL+C");  
       		  //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）  
              QueueingConsumer.Delivery delivery = consumer.nextDelivery(1000);
              if(delivery!=null)
              {
            	  AckResult result=AckResult.RETRY;
            	  try {
            		  String message = new String(delivery.getBody());  
                	  System.out.println(" [x] Received '" + message + "'");  
                	  Thread.sleep(1000);
                	  result=AckResult.ACCEPT;
				} catch (Exception e) {
					// TODO: handle exception
				}finally{
					//basicAck方法通知服务端，如果没有调用，消息将会进入unacknowledged状态，并且当消费者连接断开后变成ready状态重新进入队列
					switch (result) {
					case ACCEPT:
						//手动返回确认状态  参数：消息的标识,false只确认当前一个消息收到,true确认所有consumer获得的消息
	              		channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 
						break;
					case RETRY:
						//ack返回false，并重新回到队列,false只确认当前一个消息收到,true确认所有consumer获得的消息
	              		channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
						break;
					case REJECT:
					default:
						//拒绝消息BasicReject一次只能拒绝接收一个消息
	              		channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true);
						break;
					}
				}
              }
			} catch (Exception e) {
				System.out.println(e.getMessage());
			}
       }
    }
    
    public static void main(String[] argv) throws Exception
    {
    	new Recv().start();
    } 
    
}  